package cn.doitedu.dmp

import org.apache.commons.lang3.StringUtils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession

/**
 * DSP日志预处理
 */
object DSPLogPreprocess {

  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.WARN)

    val spark = SparkSession.builder()
      .appName("DMP日志用户标识IDMAPPING映射字典生成")
      .master("local")
      .getOrCreate()

    import spark.implicits._


    // 加载日志文件
    val ds = spark.read.textFile("user_portrait/data/dmp_log")

    // 加载今天做好的idmapping字典
    val idmp = spark.read.parquet("user_portrait/data/dmp_log_idmpdict")
    val idmp_map = idmp.rdd.map(row => {
      val flagId = row.getAs[Long](0)
      val guid = row.getAs[Long](1)
      (flagId, guid)
    }).collectAsMap()
    val bc = spark.sparkContext.broadcast(idmp_map)

    // 加载geohash字典
    // TODO

    // 加载app信息字典
    // TODO


    // 开始日志数据的处理逻辑
    val res = ds
      .map(line => {
        line.split(",")
      })
      .filter(_.size >= 85)
      .filter(arr => {
        val x = arr.slice(46, 51).filter(StringUtils.isNotBlank(_))
        x.size > 0
      })
      // 还可以过滤一些关键字段非法的数据： 请求类型，竞价价格等
      .mapPartitions(iter => {
        val idmpDict = bc.value
        iter.map(arr => {
          val logBean = BeanUtil.arr2Bean(arr)
          // 取到gps坐标，关联geohash地理位置字典
          // 取到appid ，关联app信息字典
          // 取到标识码，关联idmapping字典得出guid
          val flag = Array(logBean.imei, logBean.idfa, logBean.mac, logBean.openudid, logBean.androidid).filter(StringUtils.isNotBlank(_))(0)
          val guid = idmpDict.get(flag.hashCode).get


          logBean.guid = guid

          logBean
        })


      }).toDF()


    // 保存到hive的分区目录中
    res.write.parquet("user_portrait/data/dmp_log_output/2020-09-01")

    spark.close()

  }
}
